package org.fb.utils;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;


/**
 * @author fanbo@imoran.net
 * @date 2021/2/22 19:51
 */
public class ZkClient {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private CuratorFramework client;

    private String zookeeperServer = "60.205.58.83:2181";

    private int sessionTimeoutMs = 6000;

    private int connectionTimeoutMs = 6000;

    private int baseSleepTimeMs = 1000;

    private int maxRetries = 3;

    public void setZookeeperServer(String zookeeperServer) {
        this.zookeeperServer = zookeeperServer;
    }

    public String getZookeeperServer() {
        return zookeeperServer;
    }

    public void setSessionTimeoutMs(int sessionTimeoutMs) {
        this.sessionTimeoutMs = sessionTimeoutMs;
    }

    public int getSessionTimeoutMs() {
        return sessionTimeoutMs;
    }

    public void setConnectionTimeoutMs(int connectionTimeoutMs) {
        this.connectionTimeoutMs = connectionTimeoutMs;
    }

    public int getConnectionTimeoutMs() {
        return connectionTimeoutMs;
    }

    public void setBaseSleepTimeMs(int baseSleepTimeMs) {
        this.baseSleepTimeMs = baseSleepTimeMs;
    }

    public int getBaseSleepTimeMs() {
        return baseSleepTimeMs;
    }

    public void setMaxRetries(int maxRetries) {
        this.maxRetries = maxRetries;
    }

    public int getMaxRetries() {
        return maxRetries;
    }

    public void init() {

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries);
        client = CuratorFrameworkFactory
                .builder()
                .connectString(zookeeperServer)
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(sessionTimeoutMs)
                .connectionTimeoutMs(connectionTimeoutMs).build();
        client.start();
    }

    public void stop() {
        client.close();
    }

    public CuratorFramework getClient() {
        return client;
    }

    /**
     * 注册接口
     *
     * @param port
     */
    public void register(String node, String hostAddress, Integer port) {
        try {
            String rootPath = "/" + "services";
            String serviceInstance = hostAddress + ":" + port;
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(rootPath + "/" + node, serviceInstance.getBytes());
        } catch (Exception e) {
            logger.error("注册出错", e);
        }
    }

    public List<String> getChildren(String path) {
        List<String> childrenList = new ArrayList<>();
        try {
            childrenList = client.getChildren().forPath(path);
        } catch (Exception e) {
            logger.error("获取子节点出错", e);
        }
        return childrenList;
    }

    /**
     * 检查Zookeeper节点是否存在
     *
     * @param nodePath 节点路径
     * @return boolean 如果存在则返回true
     */
    public boolean checkExists(String nodePath) {
        try {
            Stat stat = client.checkExists().forPath(nodePath);
            return stat != null;
        } catch (Exception e) {
            logger.error(MessageFormat.format("检查Zookeeper节点是否存在出现异常,nodePath:{0}", nodePath));
        }
        return false;
    }

    /**
     * 获取某个Zookeeper节点的数据
     *
     * @param nodePath 节点路径
     * @return java.lang.String
     */
    public String getData(String nodePath) {
        try {
            return new String(client.getData().forPath(nodePath));
        } catch (Exception e) {
            logger.error(MessageFormat.format("获取某个Zookeeper节点的数据出现异常,nodePath:{0}", nodePath), e);
        }
        return null;
    }


    /**
     * 级联删除某个Zookeeper节点及其子节点
     *
     * @param nodePath 节点路径
     */
    public void deleteChildrenIfNeeded(String nodePath) {
        try {
            client.delete().guaranteed().deletingChildrenIfNeeded().forPath(nodePath);
        } catch (Exception e) {
            logger.error(MessageFormat.format("级联删除某个Zookeeper节点及其子节点出现异常,nodePath:{0}", nodePath), e);
        }
    }

    /**
     * 创建永久Zookeeper节点
     *
     * @param nodePath  节点路径（如果父节点不存在则会自动创建父节点），如：/curator
     * @param nodeValue 节点数据
     * @return java.lang.String 返回创建成功的节点路径
     */
    public String createPersistentNode(String nodePath, String nodeValue) {
        try {
            if(nodeValue != null && !nodeValue.equals("")) {
                return client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(nodePath, nodeValue.getBytes());
            }else {
                return client.create().creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(nodePath);
            }
        } catch (Exception e) {
            logger.error(MessageFormat.format("创建永久Zookeeper节点失败, nodePath:{0}, nodeValue:{1}", nodePath, nodeValue), e);
        }
        return null;
    }



    /**
     * 删除某个Zookeeper节点
     *
     * @param nodePath 节点路径
     */
    public void delete(String nodePath) {
        try {
            client.delete().guaranteed().forPath(nodePath);
        } catch (Exception e) {
            logger.error(MessageFormat.format("删除某个Zookeeper节点出现异常,nodePath:{0}", nodePath), e);
        }
    }

    public int getChildrenCount(String path) {
        return getChildren(path).size();
    }

    public List<String> getInstances() {
        return getChildren("/services");
    }

    public int getInstancesCount() {
        return getInstances().size();
    }
}
